package com.hbq.codedemopersion.hbase.impl;

import com.hbq.codedemopersion.hbase.HbaseCreateTableDTO;
import com.hbq.codedemopersion.hbase.HbaseDataDTO;
import com.hbq.codedemopersion.hbase.HbasePutDTO;
import com.hbq.codedemopersion.hbase.IHBaseService;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author: huibq
 * @Date: 2022/10/26 15:07
 * @Description: HBaseServiceImpl
 */
//@Component
@Slf4j
public class HBaseServiceImpl implements IHBaseService {
    @Resource(name = "hBaseConnection")
    private Connection hBaseConnection;
    @Resource(name = "hBaseAdmin")
    private Admin hBaseAdmin;

    /**
     * 根据rowKey查询
     * 不推荐使用，具体可以切换到 详细Map查询
     *
     * @param tableName 表名
     * @param rowKeys   名称
     * @param columns   列名
     * @return 返回对象
     */
    @Deprecated
    public Map<String, Map<String, String>> queryByRowKey(String tableName, List<String> rowKeys, Set<String> columns, String family) {
        Map<String, Set<String>> rowKeyColumn = new HashMap<>(rowKeys.size());
        rowKeys.forEach(rowKey -> columns.forEach(column -> rowKeyColumn.computeIfAbsent(rowKey, k -> new HashSet<>()).add(column)));
        return queryByRowKey(tableName, rowKeyColumn, family);
    }

    /**
     * 按map分类查询rowKey
     *
     * @param tableName    表名
     * @param rowKeyColumn 对应map关系
     * @param family       列族
     * @return 结果
     */
    public Map<String, Map<String, String>> queryByRowKey(String tableName, Map<String, Set<String>> rowKeyColumn, String family) {
        Map<String, Map<String, String>> result = new HashMap<>(32);
        try (Table table = hBaseConnection.getTable(TableName.valueOf(tableName))) {
            List<Get> gets = calculateGets(rowKeyColumn, family);
            Result[] resArr = table.get(gets);
            for (Result rst : resArr) {
                Cell[] cells = rst.rawCells();
                Map<String, String> cv = new HashMap<>(cells.length);
                for (Cell cell : cells) {
                    String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    cv.put(colName, value);
                }
                if (cv.size() != 0) {
                    result.put(Bytes.toString(rst.getRow()), cv);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("查询指定表中的所有数据失败,tableName:{}，e:{}", tableName, e.getMessage());
        }
        return result;
    }

    /**
     * 按map分类查询rowKey
     *
     * @param tableName    表名
     * @param rowKeyColumn 对应map关系
     * @param family       列族
     * @param callback     回调函数
     */
    public <R> void processBatchCallback(String tableName, Map<String, Set<String>> rowKeyColumn, String family, final Batch.Callback<R> callback) {
        try (Table table = hBaseConnection.getTable(TableName.valueOf(tableName))) {
            List<Get> gets = calculateGets(rowKeyColumn, family);
            Object[] r1 = new Object[gets.size()];
            table.batchCallback(gets, r1, callback);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
            log.error("查询指定表中的所有数据失败,tableName:{}，e:{}", tableName, e.getMessage());
        }
    }


    private List<Get> calculateGets(Map<String, Set<String>> rowKeyColumn, String family) {
        List<Get> gets = new ArrayList<>(rowKeyColumn.size());
        for (Map.Entry<String, Set<String>> entry : rowKeyColumn.entrySet()) {
            String key = entry.getKey();
            Set<String> columns = entry.getValue();
            gets.add(createGet(key, columns, family));
        }
        return gets;
    }

    private Get createGet(String key, Set<String> columns, String family) {
        Get get = new Get(key.getBytes());
        if (!get.isCheckExistenceOnly()) {
            for (String col : columns) {
                get.addColumn(Bytes.toBytes(family), Bytes.toBytes(col));
            }
            get.setCacheBlocks(false);
        }
        return get;
    }


    /**
     * 按rowKey前缀查询，不指定列，会拿到family中全部cell
     *
     * @param tableName    表名
     * @param rowKeyPrefix rowKey前缀
     * @param family       列族
     * @return 结果
     */
    public List<Map<String, String>> queryByRowKeyPrefix(String tableName, String rowKeyPrefix, String family) {
        List<Map<String, String>> result = new ArrayList<>(32);
        try (Table table = hBaseConnection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            PrefixFilter filter = new PrefixFilter(Bytes.toBytes(rowKeyPrefix));
            scan.setFilter(filter);
            scan.addFamily(Bytes.toBytes(family));
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result rs : resultScanner) {
                // hbase的一个row对应的所有cell
                List<Cell> cells = rs.listCells();
                Map<String, String> cv = new HashMap<>(cells.size());
                for (Cell cell : cells) {//value
                    String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    cv.put(colName, value);
                }
                cv.put("rowKey", Bytes.toString(rs.getRow()));
                if (cv.size() != 0) {
                    result.add(cv);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("查询指定表中的所有数据失败,tableName:{}，e:{}", tableName, e.getMessage());
        }
        return result;
    }

    @Override
    public void putData(String tableName, Map<String, Map<String, String>> rowKeyColumn, String family) throws IOException {
        //获取表对象
        Table table = hBaseConnection.getTable(TableName.valueOf(tableName));
        rowKeyColumn.entrySet().forEach(
                entry -> {
                    String rowKey = entry.getKey();
                    Map<String, String> value = entry.getValue();
                    Put put = new Put(Bytes.toBytes(rowKey));
                    value.entrySet().forEach(
                            column -> {
                                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column.getKey()), Bytes.toBytes(column.getValue()));
                                log.info("【put table data:{},rowKey:{},family:{},columnKey:{},columnValue:{}】", tableName, rowKey, family, column.getKey(), column.getValue());
                            }
                    );
                    try {
                        table.put(put);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
        );
        table.close();
        log.info("插入数据成功");
    }

    public void putData(HbasePutDTO hbasePutDto) throws IOException {
        HbasePutDTO.RowKeyColumn rowKeyColumn = hbasePutDto.getRowKeyColumn();
        Map<String, String> columnMap = new HashMap<>();
        rowKeyColumn.getColumn().forEach(
                column -> {
                    columnMap.put(column.getName(),column.getValue());
                }
        );
        this.putData(hbasePutDto.getTableName(),Map.of(rowKeyColumn.getRowKey(),columnMap) , hbasePutDto.getFamily());
    }

    @Override
    public void createTable(String tableName, Set<String> familyList) throws IOException {
        //创建列族
        List<ColumnFamilyDescriptor> columnFamilyDescriptorList = familyList.stream()
                .map(family -> ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build())
                .collect(Collectors.toList());
        TableDescriptor desc = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))//指定表名
                .setColumnFamilies(columnFamilyDescriptorList)
                .build();
        //创建表
        hBaseAdmin.createTable(desc);
        //关闭连接
        hBaseAdmin.close();
        log.info("创建【table:{},familyList:{}】成功", tableName, familyList);
    }

    public void removeTable(String tableName) throws IOException {
        //删除表，先禁用表
        hBaseAdmin.disableTable(TableName.valueOf(tableName));
        hBaseAdmin.deleteTable(TableName.valueOf(tableName));
    }

    public void deleteData(String tableName, Map<String, Set<String>> rowKeyColumn, String family) throws IOException {
        //获取Table，指定要操作的表名，表需要提前创建好
        Table table = hBaseConnection.getTable(TableName.valueOf(tableName));
        rowKeyColumn.entrySet().forEach(
                entry -> {
                    String rowKey = entry.getKey();
                    Set<String> value = entry.getValue();
                    Delete delete = new Delete(Bytes.toBytes(rowKey));
                    value.forEach(
                            column -> {
                                delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
                                log.info("【delete table data:{},rowKey:{},family:{},column:{}】", tableName, rowKey, family, column);
                            }
                    );
                    try {
                        table.delete(delete);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
        );
    }

    @Override
    public void createTable(HbaseCreateTableDTO hbaseCreateTableDto) throws IOException {
        createTable(hbaseCreateTableDto.getTableName(), hbaseCreateTableDto.getFamilySet());
    }

    @Override
    public void deleteData(HbaseDataDTO hbaseDataDTO) {
        try {
            deleteData(hbaseDataDTO.getTableName(),
                    Map.of(hbaseDataDTO.getRowKeyColumn().getRowKey(), hbaseDataDTO.getRowKeyColumn().getColumnName()),
                    hbaseDataDTO.getFamily());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Map<String, Map<String, String>> queryByRowKey(HbaseDataDTO hbaseDataDTO) {
        return queryByRowKey(hbaseDataDTO.getTableName(),
                Map.of(hbaseDataDTO.getRowKeyColumn().getRowKey(),hbaseDataDTO.getRowKeyColumn().getColumnName()),
                hbaseDataDTO.getFamily());
    }

    public List<String> queryByRowKey(String tableName, String rowKey, String family) {
        List<String> result = new ArrayList<>(32);
        Get get = new Get(rowKey.getBytes());
        get.addFamily(family.getBytes());
        try (Table table = hBaseConnection.getTable(TableName.valueOf(tableName))) {
            Result resArr = table.get(get);
            Cell[] cells = resArr.rawCells();
            for (Cell cell : cells) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                result.add(colName);
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("查询指定表中的所有数据失败,tableName:{}，e:{}", tableName, e.getMessage());
        }
        return result;
    }
}
